package cn.me.implement.threadpool.v4_0;

import cn.me.implement.threadpool.v4_0.reject.RejectPolicy;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <h3>线程池 4.0 版本</h3>
 * 它实现的功能如下：
 * <ul>
 *     <li>提供了关闭线程池的方法</li>
 *     <li>可以用一个工具类来创建线程池</li>
 *     <li>
 *         提供给使用者多个构造方法，最少指定五个参数，最多指定七个参数：
 *         <ul>
 *             <li><strong>核心线程数量</strong></li>
 *             <li><strong>最大线程数量</strong></li>
 *             <li><strong>临时线程在等待领取任务时最大阻塞时间</strong></li>
 *             <li><strong>阻塞时间的单位</strong></li>
 *             <li><strong>使用者指定的阻塞队列，充当存放任务的任务队列</strong></li>
 *             <li><strong>拒绝策略，默认为抛出异常的拒绝策略</strong></li>
 *             <li><strong>线程工厂，创建线程时使用，默认为线程添加了简单的名字</strong></li>
 *         </ul>
 *     </li>
 *     <li>在无法执行任务的情况下提供了拒绝策略</li>
 *     <li>使用锁来保证 {@link #threadPool} 相关操作的 <strong>互斥性</strong></li>
 *     <li>
 *         将线程分为两类：
 *         <ul>
 *             <li><strong>核心线程</strong>：永不停止</li>
 *             <li><strong>临时线程</strong>：在一段时间内没有获取任务的情况下停止</li>
 *         </ul>
 *     </li>
 *     <li>使用一个 <strong>{@link Set}</strong> 存放所有线程</li>
 *     <li><strong>在新建线程时交给线程一个任务</strong></li>
 *     <li><strong>线程在提交任务时才创建</strong></li>
 *     <li><strong>{@link #submit(Runnable task)} 通过任务队列提交任务，线程通过任务队列领取任务</strong></li>
 * </ul>
 */
public class ThreadPool4_0 {

    /**
     * 线程池中核心线程的最大数量
     */
    private final int corePoolSize;

    /**
     * 线程池中线程的最大数量
     */
    private final int maxPoolSize;

    /**
     * 临时线程阻塞的最长时间（单位：ns），超过这个时间还没有领取到任务就直接退出
     */
    private final long keepAliveTime;

    /**
     * 任务队列
     */
    private final BlockingQueue<Runnable> taskQueue;

    /**
     * 拒绝策略，用于在无法执行任务的时候拒绝任务
     */
    private final RejectPolicy rejectPolicy;

    /**
     * 线程工厂
     */
    private final ThreadFactory threadFactory;

    /**
     * 默认的线程工厂
     */
    private static final ThreadFactory DEFAULT_THREAD_FACTORY = new ThreadFactory() {

        /**
         * 计数器，用来记录当前创建的是第几个线程，从 0 开始
         */
        private int counter = 0;

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "thread-pool-" + counter++);
        }
    };

    /**
     * 构造一个线程池，默认参数如下：
     * <ul>
     *     <li>拒绝策略默认为抛出异常的拒绝策略</li>
     *     <li>线程工厂默认为线程添加了简单的名字 thread-pool-?</li>
     * </ul>
     *
     * @param corePoolSize 线程池中核心线程的最大数量
     * @param maxPoolSize 线程池中线程的最大数量
     * @param keepAliveTime 临时线程阻塞的最长时间
     * @param unit 时间的单位
     * @param taskQueue 任务队列
     */
    public ThreadPool4_0(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> taskQueue) {
        this(corePoolSize, maxPoolSize, keepAliveTime, unit, taskQueue, RejectPolicy.THROW_EXCEPTION);
    }

    /**
     * 构造一个线程池，默认参数如下：
     * <ul>
     *     <li>拒绝策略默认为抛出异常的拒绝策略</li>
     * </ul>
     *
     * @param corePoolSize 线程池中核心线程的最大数量
     * @param maxPoolSize 线程池中线程的最大数量
     * @param keepAliveTime 临时线程阻塞的最长时间
     * @param unit 时间的单位
     * @param taskQueue 任务队列
     * @param threadFactory 线程工厂
     */
    public ThreadPool4_0(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> taskQueue, ThreadFactory threadFactory) {
        this(corePoolSize, maxPoolSize, keepAliveTime, unit, taskQueue, RejectPolicy.THROW_EXCEPTION, threadFactory);
    }

    /**
     * 构造一个线程池，默认参数如下：
     * <ul>
     *     <li>线程工厂默认为线程添加了简单的名字 thread-pool-?</li>
     * </ul>
     *
     * @param corePoolSize 线程池中核心线程的最大数量
     * @param maxPoolSize 线程池中线程的最大数量
     * @param keepAliveTime 临时线程阻塞的最长时间
     * @param unit 时间的单位
     * @param taskQueue 任务队列
     * @param rejectPolicy 拒绝策略
     */
    public ThreadPool4_0(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> taskQueue, RejectPolicy rejectPolicy) {
        this(corePoolSize, maxPoolSize, keepAliveTime, unit, taskQueue, rejectPolicy, DEFAULT_THREAD_FACTORY);
    }

    /**
     * 构造一个线程池
     *
     * @param corePoolSize 线程池中核心线程的最大数量
     * @param maxPoolSize 线程池中线程的最大数量
     * @param keepAliveTime 临时线程阻塞的最长时间
     * @param unit 时间的单位
     * @param taskQueue 任务队列
     * @param rejectPolicy 拒绝策略
     * @param threadFactory 线程工厂
     */
    public ThreadPool4_0(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> taskQueue, RejectPolicy rejectPolicy, ThreadFactory threadFactory) {
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.taskQueue = taskQueue;
        this.rejectPolicy = rejectPolicy;
        this.threadFactory = threadFactory;
    }

    /**
     * 存放线程的集合，使用 {@link Set} 是因为 {@link Set#remove(Object)} 性能更高
     */
    private final Set<Worker> threadPool = new HashSet<>();

    /**
     * 线程池的管程
     * <p>
     * 用于保证 <strong>将线程放入线程池</strong>、<strong>从线程池中移除线程</strong> 的互斥性
     * 同时也在保证 {@link #currPoolSize} 相关操作的互斥性
     */
    private final Object threadPoolMonitor = new Object();

    /**
     * 线程池中当前线程数量，这个值 <= threadPool.size()
     * 在创建新线程时增加，在放毒丸时减少，threadPool.size() 减少的时机晚于 currPoolSize
     */
    private int currPoolSize = 0;

    /**
     * <h3>核心线程执行的任务</h3>
     * {@link #getTask()} 方法会一直阻塞，直到有新任务
     */
    public final class CoreWorker extends Worker {

        public CoreWorker(Runnable initialTask, Set<Worker> workerPool, ThreadFactory threadFactory,
                ThreadPool4_0 threadPool) {
            super(initialTask, workerPool, threadFactory, threadPool);
        }

        @Override
        protected Runnable getTask() {
            try {
                return taskQueue.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * <h3>临时线程执行的任务</h3>
     * {@link #getTask()} 方法会在阻塞一定时间后如果还没有任务，则会返回 {@code null}
     */
    public final class TempWorker extends Worker {

        public TempWorker(Runnable initialTask, Set<Worker> workerPool, ThreadFactory threadFactory,
                ThreadPool4_0 threadPool) {
            super(initialTask, workerPool, threadFactory, threadPool);
        }

        @Override
        protected Runnable getTask() {
            try {
                return taskQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 线程池的状态，状态共有 2 种：
     * <ul>
     *     <li>{@link #RUNNING} 运行状态</li>
     *     <li>{@link #SHUTDOWN} 关闭状态，调用了线程池的关闭方法</li>
     * </ul>
     */
    private final AtomicInteger state = new AtomicInteger(RUNNING);

    private static final int RUNNING = 1;
    private static final int SHUTDOWN = 2;

    /**
     * 提交任务
     *
     * @param task 待执行的任务
     */
    public void submit(Runnable task) {
        // 如果线程池的状态不是 RUNNING 状态，则直接拒绝任务
        if (state.get() != RUNNING) {
            rejectPolicy.reject(this, task);
            return;
        }

        // 如果 线程数量 小于 最大核心线程数量，则新建一个 核心线程 执行任务，然后直接返回
        synchronized (threadPoolMonitor) {
            if (currPoolSize < corePoolSize) {
                CoreWorker coreWorker = new CoreWorker(task, threadPool, threadFactory, this);
                coreWorker.start();
                currPoolSize++;
                return;
            }
        }

        // 如果能够放到任务队列中，则直接返回
        if (taskQueue.offer(task)) {
            return;
        }

        // 如果 线程数量 小于 最大线程数量，则新建一个 临时线程 执行任务
        synchronized (threadPoolMonitor) {
            if (currPoolSize < maxPoolSize) {
                TempWorker tempWorker = new TempWorker(task, threadPool, threadFactory, this);
                tempWorker.start();
                currPoolSize++;
                return;
            }
        }

        // 线程数量到达最大线程数量，任务队列已满，执行拒绝策略
        rejectPolicy.reject(this, task);
    }

    /**
     * 在没有任务执行时停止所有线程
     * 当此方法被调用，线程池会停止接受提交任务，然后等待线程将 它们正在执行的任务 和 任务队列中的任务 都执行完毕，之后让所有线程退出
     */
    public void shutdownWhenNoTask() {
        // 将状态从 RUNNING 切换到 SHUTDOWN
        if (state.compareAndSet(RUNNING, SHUTDOWN)) {
            // 如果切换成功，则向任务队列中投放一个毒丸
            offerPoisonPill();
        }
    }

    /**
     * 立刻停止所有线程
     * 当此方法被调用，线程池会停止接受提交任务，给线程发送中断信号
     * <p>
     * 注意：<strong>如果在调用此方法之前调用了 {@link #shutdownWhenNoTask()} 方法，不会立刻停止所有线程</strong>
     *
     * @return 任务队列中的任务
     */
    public List<Runnable> shutdownNow() {
        // 将状态从 RUNNING 切换到 SHUTDOWN，如果修改失败，则表示线程池已经调用过关闭相关的方法了，直接返回一个空集合即可
        if (!state.compareAndSet(RUNNING, SHUTDOWN)) {
            return new ArrayList<>();
        }

        // 给所有线程发送中断信号
        synchronized (threadPoolMonitor) {
            threadPool.forEach(Worker::interrupt);
        }
        // 将任务队列中的任务放到一个集合中
        List<Runnable> taskList = new ArrayList<>(taskQueue.size());
        taskQueue.drainTo(taskList);
        // 向任务队列中投放一个毒丸
        offerPoisonPill();
        // 返回任务集合
        return taskList;
    }

    /**
     * 获取当前线程池中的线程数量
     *
     * @return 当前线程池中的线程数量
     */
    public int getCurrPoolSize() {
        synchronized (threadPoolMonitor) {
            return currPoolSize;
        }
    }

    /**
     * 获取当前任务队列中的任务数
     *
     * @return 当前任务队列中的任务数
     */
    public int getCurrTaskNum() {
        return taskQueue.size();
    }

    /**
     * 丢弃任务队列 {@link #taskQueue} 中的最旧的任务（队头任务）
     *
     * @return 任务队列中的最旧的任务（队头任务）
     */
    public Runnable discardOldestTask() {
        return taskQueue.poll();
    }

    /**
     * 毒丸，用于让线程池中的所有线程退出
     * 投放到任务队列中，只要线程获取到这个任务，就退出，并且在 线程池中还有线程 的情况下将毒丸重新放回任务队列
     */
    public static final Runnable POISON_PILL = () -> {
    };

    // 实际上投放毒丸的操作
    private void doOfferPoisonPill() {
        // 投放毒丸，直到成功为止
        try {
            taskQueue.put(POISON_PILL);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        currPoolSize--;
    }

    /**
     * 向任务队列中投放一个毒丸，等待线程领取后退出
     */
    public void offerPoisonPill() {
        synchronized (threadPoolMonitor) {
            doOfferPoisonPill();
        }
    }

    /**
     * 在线程池中还有其他线程的情况下，向任务队列中投放一个毒丸，等待线程领取后退出，用于
     */
    public void offerPoisonPillIfThreadRest() {
        synchronized (threadPoolMonitor) {
            if (currPoolSize > 0) {
                doOfferPoisonPill();
            }
        }
    }

    /**
     * 从 {@link #threadPool} 中移除指定的 {@link Worker} 对象
     *
     * @param worker 待移除的 {@link Worker} 对象
     */
    public void removeWorkerFromThreadPool(Worker worker) {
        synchronized (threadPoolMonitor) {
            threadPool.remove(worker);
        }
    }
}
